package cn.itcast.flink.base;

import org.apache.flink.api.common.serialization.SimpleStringEncoder;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.core.fs.Path;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.filesystem.OutputFileConfig;
import org.apache.flink.streaming.api.functions.sink.filesystem.StreamingFileSink;
import org.apache.flink.streaming.api.functions.sink.filesystem.bucketassigners.DateTimeBucketAssigner;
import org.apache.flink.streaming.api.functions.sink.filesystem.rollingpolicies.DefaultRollingPolicy;

/**
 * Author itcast
 * Date 2021/7/31 11:34
 * Desc TODO
 */
public class StreamingFileSinkDemo {
    public static void main(String[] args) throws Exception {
        //设置参数
        ParameterTool tool = ParameterTool.fromArgs(args);
        //设置 hostname 和 port
        String hostname = tool.getRequired("hostname");
        int port = tool.getInt("port");
        //1．初始化流计算运行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //2．设置Checkpoint（10s）周期性启动 和 stateBackend 存储路径
        env.enableCheckpointing(10000);
        env.setStateBackend(new FsStateBackend("file:///d:/chk"));
        //3．指定并行度为1
        env.setParallelism(1);
        //4．接入socket数据源，获取数据
        DataStreamSource<String> source = env.socketTextStream(hostname, port);
        //5．创建Streamingfilesink对象
        //5-1. 创建输出文件配置，指定输出路径 /FlinkStreamFileSink/parquet
        OutputFileConfig config = OutputFileConfig
                .builder()
                .withPartPrefix("prefix")
                .withPartSuffix(".txt")
                .build();
        //设置 StreamingFileSink
        StreamingFileSink<String> fileSink = StreamingFileSink
                .forRowFormat(
                new Path("/FlinkStreamFileSink/parquet"),
                new SimpleStringEncoder()
                ).withBucketAssigner(new DateTimeBucketAssigner())
                .withRollingPolicy(DefaultRollingPolicy.builder()
                .withRolloverInterval(Time.seconds(2).toMilliseconds())
                .withInactivityInterval(Time.seconds(1).toMilliseconds())
                .withMaxPartSize(1024*1024*128)
                .build())
                .withOutputFileConfig(config)
                .build();
        //5-2.StreamingFileSink 行格式化 ， withBucketAssigner->DateTimeBucketAssigner
        //withRollingPolicy -> 默认滚筒策略
        //withOutputFileConfig -> 输出文件的配置

        //6.设置输出 sink
        source.addSink(fileSink).setParallelism(1);

        //7．执行任务
        env.execute();
    }
}
